package bma.common.netty.thrift.dual;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.thrift.TApplicationException;
import org.apache.thrift.TBase;
import org.apache.thrift.protocol.TMessage;
import org.apache.thrift.protocol.TMessageType;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TIOStreamTransport;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBufferInputStream;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;

import bma.common.langutil.ai.AIUtil;
import bma.common.langutil.ai.common.AITimeoutException;
import bma.common.langutil.ai.stack.AIStack;
import bma.common.langutil.ai.stack.AIStackROOT;
import bma.common.netty.thrift.core.NettyThriftConfig;
import bma.common.netty.thrift.core.NettyThriftTimeout.Timeout;
import bma.common.netty.thrift.core.TNettyFramedWriteOnlyTransport;
import bma.common.netty.thrift.dual.impl.NTSRequestImpl;
import bma.common.netty.thrift.dual.impl.NTSResponseImpl;

/**
 * 处理Protobuf请求
 * 
 * @author guanzhong
 * 
 */
public class NCHThriftService extends SimpleChannelUpstreamHandler {

	final org.slf4j.Logger log = org.slf4j.LoggerFactory
			.getLogger(NCHThriftService.class);

	protected NettyThriftConfig config;
	protected NTSProcessor processor;

	protected AtomicInteger seqId = new AtomicInteger();
	protected Map<Integer, AIStack<NTSResponse>> await = new ConcurrentHashMap<Integer, AIStack<NTSResponse>>();

	public NCHThriftService() {
		super();
	}

	public NCHThriftService(NettyThriftConfig cfg, NTSProcessor p) {
		super();
		this.config = cfg;
		this.processor = p;
	}

	public NTSProcessor getProcessor() {
		return this.processor;
	}

	public void setProcessor(NTSProcessor p) {
		this.processor = p;
	}

	public NettyThriftConfig getConfig() {
		return config;
	}

	public void setConfig(NettyThriftConfig config) {
		this.config = config;
	}

	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
			throws Exception {
		Object obj = e.getMessage();
		if (!(obj instanceof ChannelBuffer)) {
			ctx.sendUpstream(e);
			return;
		}

		ChannelBuffer buffer = (ChannelBuffer) obj;
		ChannelBufferInputStream in = new ChannelBufferInputStream(buffer);
		TProtocol inpro = config.getProtocol(new TIOStreamTransport(in));

		TMessage msg = inpro.readMessageBegin();

		if (log.isDebugEnabled()) {
			log.debug(msg.toString());
		}

		switch (msg.type) {
		case TMessageType.CALL:
		case TMessageType.ONEWAY: {
			NTSRequestImpl req = new NTSRequestImpl();
			req.setChannel(ctx.getChannel());
			req.setConfig(config);
			req.setInputProtocol(inpro);
			req.setMessage(msg);
			processRequest(msg, req);
			break;
		}
		case TMessageType.EXCEPTION: {
			TApplicationException x = TApplicationException.read(inpro);
			processResponse(msg, null, x);
			inpro.readMessageEnd();
			break;
		}
		case TMessageType.REPLY: {
			NTSResponseImpl resp = new NTSResponseImpl();
			resp.setChannel(ctx.getChannel());
			resp.setConfig(config);
			resp.setInputProtocol(inpro);
			resp.setMessage(msg);
			processResponse(msg, resp, null);
			break;
		}
		default:
			log.warn("unknow message type[{}], close channel", msg.type);
			ctx.getChannel().close();
			break;
		}
	}

	private void processResponse(TMessage msg, final NTSResponseImpl resp,
			final Exception err) {
		// response or error
		int id = msg.seqid;
		final AIStack<NTSResponse> stack = await.remove(id);
		if (stack != null) {
			config.getTimeout().remove(resp.getChannel().getId(), id);
		}
		if (stack != null) {
			config.getExecutor().execute(new Runnable() {

				@Override
				public void run() {
					if (err != null) {
						AIUtil.safeFailure(stack, err);
					} else {
						AIUtil.safeSuccess(stack, resp);
					}
				}
			});

		} else {
			if (log.isDebugEnabled()) {
				log.debug("skip response - [{}]", id);
			}
		}
	}

	private void processRequest(final TMessage msg, final NTSRequestImpl req) {
		AIStack<NTSProcessor> stack = new AIStackROOT<NTSProcessor>() {

			@Override
			public boolean end(final NTSProcessor result, Throwable t) {
				if (t != null) {
					req.end(null, t);
				} else {
					if (result == null) {
						if (log.isWarnEnabled()) {
							log.warn("'{}' processor not found", msg.name);
						}
						req.end(null, new TApplicationException(
								TApplicationException.UNKNOWN_METHOD, "'"
										+ msg.name + "' miss processor"));
					} else {
						config.getExecutor().execute(new Runnable() {

							@Override
							public void run() {
								try {
									result.process(req);
								} catch (Exception e) {
									if (log.isWarnEnabled()) {
										log.warn("handle service error - "
												+ msg.name, e);
									}
									req.end(null, e);
								}
							}
						});

					}
				}

				return true;
			}

		};
		if (processor instanceof NTSDispatcher) {
			NTSDispatcher dispatcher = (NTSDispatcher) processor;
			dispatcher.dispatch(stack, msg.name, req);
		} else {
			stack.success(processor);
		}

	}

	@Override
	public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)
			throws Exception {
		super.channelClosed(ctx, e);
		List<Map.Entry<Integer, AIStack<NTSResponse>>> tmp = new ArrayList<Map.Entry<Integer, AIStack<NTSResponse>>>(
				await.entrySet());
		await.clear();
		int cid = e.getChannel().getId();
		Throwable t = new IOException("closed");
		for (Map.Entry<Integer, AIStack<NTSResponse>> item : tmp) {
			config.getTimeout().remove(cid, item.getKey());
			AIUtil.safeFailure(item.getValue(), t);
		}
	}

	public int sendRequest(Channel ch, AIStack<NTSResponse> stack, String name,
			TBase data, int timeout) {
		return sendRequest(ch, stack, name, TMessageType.CALL, data, timeout,
				false);
	}

	public int sendRequest(Channel ch, AIStack<NTSResponse> stack, String name,
			int type, TBase data, int timeout, boolean closed) {
		int seqid = 0;
		for (;;) {
			seqid = seqId.incrementAndGet();
			if (seqid == Integer.MAX_VALUE)
				seqid = 1;
			if (!await.containsKey(seqid)) {
				break;
			}
		}

		try {
			TNettyFramedWriteOnlyTransport transport = new TNettyFramedWriteOnlyTransport(
					ch, config.getMaxLength());
			TProtocol out = config.getProtocol(transport);
			out.writeMessageBegin(new TMessage(name, TMessageType.CALL, seqid));
			data.write(out);
			out.writeMessageEnd();
			out.getTransport().flush();
		} catch (Exception err) {
			AIUtil.safeFailure(stack, err);
			return seqid;
		}

		if (stack != null) {
			await.put(seqid, stack);
		}
		if (timeout > 0) {
			config.getTimeout().add(timeout, new Timeout(ch.getId(), seqid) {

				@Override
				public void timeout() {
					AIStack<NTSResponse> stack = await.remove(seqid);
					if (stack != null) {
						Throwable t = new AITimeoutException();
						AIUtil.safeFailure(stack, t);
					}
				}
			});
		}
		return seqid;
	}

	public static NTSClient getNTSClientFromChannel(Channel ch) {
		NCHThriftService service = ch.getPipeline().get(NCHThriftService.class);
		if (service == null)
			return null;
		return service.getNTSClient(ch);
	}

	protected NTSClient getNTSClient(final Channel channel) {
		return new NTSClient() {

			@Override
			public boolean isOpen() {
				return channel.isOpen();
			}

			@Override
			public void close() {
				channel.close();
			}

			@Override
			public void addCloseListener(final NTSClientCloseListener lis) {
				channel.getCloseFuture().addListener(
						new ChannelFutureListener() {

							@Override
							public void operationComplete(ChannelFuture future)
									throws Exception {
								lis.onClose();
							}
						});
			}

			@Override
			public int write(AIStack<NTSResponse> stack, String name, int type,
					TBase data, int timeout) {
				return sendRequest(channel, stack, name, data, timeout);
			}

			@Override
			public void discard(int id) {
				if (await.remove(id) != null) {
					config.getTimeout().remove(channel.getId(), id);
				}
			}
		};
	}
}
